package com.erbao.io.practice;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 管理线程 的 组
 * @author CaoBaoYin
 * @since 2022/10/10 上午9:10
 */
public class SelectorThreadGroup {
    SelectorThread[] sts;

    ServerSocketChannel server;

    // 轮循分配 SelectorThread 。从数组中轮循取出，需要取模
    AtomicInteger xid = new AtomicInteger(0);

    // 创建boss group 和 worker group区分 server注册的线程管理组 和 client注册的线程组
    SelectorThreadGroup workerGroup;

    public void setWorkerGroup(SelectorThreadGroup workerGroup) {
        this.workerGroup = workerGroup;
    }

    public SelectorThreadGroup(int num) {
        // 几个线程 SelectorThread
        sts = new SelectorThread[num];
        for (int i = 0; i < num; i++) {
            sts[i] = new SelectorThread(this);

            // 启动
            new Thread(sts[i]).start();
        }

    }

    public void bind(int port) {
        try {
            server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.bind(new InetSocketAddress(port));

            // server 注册到哪个线程上
            nextSelectorV3(server);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void nextSelectorV3(Channel channel) {
        if (channel instanceof ServerSocketChannel) {
            nextSelector(channel);
            return;
        }

        workerGroup.nextSelector(channel);
    }

    public void nextSelectorV2 (Channel channel) {
        if (channel instanceof ServerSocketChannel) {
            try {
                sts[0].lbq.put(channel);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            sts[0].selector.wakeup();
            return;
        }

        SelectorThread st = nextV2();
        st.lbq.add(channel);
        st.selector.wakeup();
    }

    /**
     * 无论ServerSocket 还是 socket 都用这个方法
     * @param channel
     */
    public void nextSelector(Channel channel) {
        SelectorThread st = next();

        // 1. 通过队列传递数据 消息
        st.lbq.add(channel);
        // 2. 打断阻塞，让对应的线程自己去注册
        st.selector.wakeup();

        /*ServerSocketChannel server = (ServerSocketChannel) channel;
        try {
            System.out.println("服务注册前……");
            *//*注册会阻塞所以需要有调用wakeup唤醒select，此处放在regiseter前后都不行，因为唤醒后无法保证哪个线程先执行
            * 解决：
            * 1.先绑定端口，selector thread 再启动线程
            * 2.队列
            * 3.线程通信*//*
            server.register(st.selector, SelectionKey.OP_ACCEPT);// 如果先执行selector.select()阻塞，这行代码也会阻塞；先执行这行代码，这行就不阻塞。
            System.out.println("服务注册后……");
        } catch (ClosedChannelException e) {
            throw new RuntimeException(e);
        }*/
    }

    /**
     * 轮循取出selector thread
     */
    private SelectorThread next () {
        int index = xid.getAndIncrement() % sts.length;
        return sts[index];
    }

    private SelectorThread nextV2 () {
        int index = xid.getAndIncrement() % (sts.length-1);//少一个轮循
        System.out.println("index: "+index);
        return sts[index+1];// 不轮循0
    }
}
